Что такое цикл опроса в Kafka

Apache Kafka является мощным инструментом для обработки потоков данных, и одной из ключевых его особенностей является способность обеспечивать высокую пропускную способность при передаче сообщений. Для эффективной обработки этих сообщений разработчики могут использовать цикл опроса, который позволяет непрерывно мониторить новые сообщения в темах Kafka.

Особенности работы цикла опроса: несколько практических примеров

Цикл опроса (Poll Loop) — это механизм, который позволяет приложению активно мониторить и обрабатывать новые данные в Kafka. В Kafka, цикл опроса обычно реализуется с использованием метода poll() внутри потока, который периодически опрашивает брокера на предмет новых сообщений [1]:

import org.apache.kafka.clients.consumer.Consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;


public class KafkaConsumerExample {

    public static void main(String[] args) {
        // Конфигурация потребителя
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");


        // Создание потребителя
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);


       // Подписка на топик
        consumer.subscribe(Collections.singletonList("my-topic"));


       // Цикл опроса
        while (true) {
            // Поллинг сообщений
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));


            // Обработка полученных записей
            records.forEach(record -> {
                System.out.printf("Получено сообщение: key = %s, value = %s%n", record.key(), record.value());
            });
        }
    }
}

В данном фрагменте кода следует отметить несколько ключевых моментов [2]:

• Конфигурация потребителя: устанавливаем необходимые параметры для настройки потребителя Kafka, такие как адрес брокера, идентификатор группы, и сериализаторы ключа и значения.
• Создание потребителя: инициализируем объект KafkaConsumer с использованием предоставленных настроек.
• Подписка на топик: указываем топик, на который хотим подписаться.
• Обработка записей: перебираем полученные записи и выполняем необходимые действия.

Здесь метод poll() в качестве параметра принимает значение, которое определяет, как часто потребитель будет проверять новые записи в топике. Группа потребителей (GROUP_ID_CONFIG) используется для организации параллельной обработки сообщений. Как видно из вышеприведенного фрагмента кода, цикл опроса представляет собой бесконечный цикл, в котором вызывается метод poll(). Этот метод возвращает записи (messages) из подписанных топиков. Стоит также отметить, что в приведенном примере мы просто выводим информацию о полученных сообщениях, но в реальном приложении здесь может быть реализована более сложная логика обработки.

Таким образом, используя цикл опроса, разработчики могут создавать эффективные приложения для обработки потоков данных в реальном времени с использованием Apache Kafka. Это делает Apache Kafka надежным и универсальным средством для хранения и обмена большими потоками данных, что активно применяется в задачах Data Science и разработке распределенных приложений.

Освоить Apache Kafka на профессиональном уровне в качестве администратора Big Data кластеров, разработчика распределенных приложений и прочих прикладных областях Data Science вы сможете на практических курсах по Kafka в нашем лицензированном учебном центре обучения и повышения квалификации ИТ-специалистов в Москве:

• Администрирование кластера Kafka
• Apache Kafka для разработчиков
• Администрирование кластера Arenadata Streaming Kafka

Записаться на курс

Смотреть раcписание

Источники

1. https://kafka.apache.org/documentation/
2. Н.Нархид, Г.Шапира, Т.Палино. Apache Kafka. Потоковая обработка и анализ данных